#include <stddef.h>  
#include <assert.h>  
#include <string.h>  
#include <iostream>
#include <unordered_map>
#include <unistd.h>
#include "redis_subscriber.h"
#include "global.h"
#include "zkprofile.h"
#include "logger.h"

using namespace std;

typedef struct  OptionRedis 
{
	char szOption[32];
	char szKey[128];
} st_OptionRedis; 

pthread_rwlock_t*             s_pMiThreadRWLock;
redisAsyncContext *           s_redisAsyContext;   
redisContext*                 s_pRedisCon;
redisReply *                  s_reply;
pthread_t                     s_pthreadEvent;



void InitLock()
{

	s_pMiThreadRWLock = (pthread_rwlock_t *)calloc(1, sizeof(pthread_rwlock_t));
	if(!s_pMiThreadRWLock)
	{
		LOG__ERROR("allocate pthread_rwlock_t for metric failed");
		return ;
	}
	pthread_rwlock_init(s_pMiThreadRWLock, NULL);
}


int getParameterValue(char* data,char* szKey,const char* szParameter)
{
	int    pos;
	char*  pEnd;
	char   szStart[128] = {0};

	pEnd= strstr(data,szParameter);
	if(!pEnd) return 0;

	pos = pEnd-szStart;
	if(pos<=0) return 0;
	strcpy(szKey,pEnd+1);

	return 1;
}

void ReadConf()
{

	FILE* fpZkUrl;
	char szline[256];  
	char szZkUrl[256];  
	char szfileName[256] = "/etc/jniRedis/conf/conf.properties";

	printf("szfileName:%s\n",szfileName);
	fpZkUrl = fopen(szfileName,"r");  
	if(fpZkUrl == NULL)  
	{         
		LOG__ERROR("zookeeper Url is NULL");
		return;  
	}
	while (fgets(szline, sizeof(szline), fpZkUrl) != NULL) 
	{
		getParameterValue(szline,szZkUrl,"=");
		if(strstr(szZkUrl,"\n"))
		{
			strncpy(g_szZookeeper,szZkUrl,strlen(szZkUrl)-1);
		}
		else
		{
			strncpy(g_szZookeeper,szZkUrl,strlen(szZkUrl));
		}
	}	


	fclose(fpZkUrl);  
}


void * event_thread(void * )
{
	event_base_dispatch(s_eventBase);

}

void start_init(void) 
{
	ReadConf();
	InitLogger();
	InitZkProfile();
	InitLock();
	//取旧数据
	connectRedis();
	getOldData();
	//监听事件
	eventInit();
	eventConnect();
	subscribe();  

	int ret = pthread_create(&s_pthreadEvent, 0, &event_thread, NULL);  
	if (ret != 0)  
	{  
		LOG__ERROR("create event thread failed.");  
	}  


}

void getOldData()
{
	p_rdisMap = new unordered_map<string,string>(1000001);
	redisReply* redisCmdReply = (redisReply*)redisCommand(s_pRedisCon, "KEYS %s",g_htProfile.m_szRedisChannel);
	redisReply* redisKey =  NULL;
	redisReply* redisGetReply = NULL;

	for(int i=0; i<redisCmdReply->elements; i++)
	{               
		redisKey = redisCmdReply->element[i];
		redisGetReply = (redisReply*)redisCommand(s_pRedisCon, "GET %s",redisKey->str);
		string strkey = redisKey->str;
		string strValue = redisGetReply->str;
		pthread_rwlock_wrlock(s_pMiThreadRWLock);
		p_rdisMap->insert(pair<string,string>(strkey,strValue));
		pthread_rwlock_unlock(s_pMiThreadRWLock);
	}
	LOG__INFO("redis get old data:%d",redisCmdReply->elements);
	freeReplyObject(redisCmdReply);

}


int connectRedis()
{
	struct timeval timeout = {2, 500000 };
	s_pRedisCon = redisConnectWithTimeout(g_htProfile.m_szRedisIp, g_htProfile.m_nPort, timeout);
	s_reply = (redisReply*)redisCommand(s_pRedisCon, "AUTH %s",g_htProfile.m_szRedisAuth);
	if(!s_pRedisCon)
	{
		LOG__ERROR("Connection error: can't allocate redis context");
		return -1;
	}

	if(s_pRedisCon->err)
	{
		LOG__ERROR("Connection error: %s", s_pRedisCon->errstr);
		redisFree(s_pRedisCon);
		freeReplyObject(s_reply);
		return -1;
	}
	LOG__INFO("redis connect is ok");
	return 0;
}

bool uninit()  
{   
	event_base_free(s_eventBase);
	eventDisconnect();
	redisFree(s_pRedisCon);	
	return true;  
} 



bool eventInit()  
{  
	s_eventBase = event_base_new();   
	if (NULL == s_eventBase)  
	{  
		LOG__ERROR("Create redis event failed.");

		return false;  
	}   
	return true;  
} 



bool eventDisconnect()  
{  
	if (s_redisAsyContext)  
	{  
		redisAsyncDisconnect(s_redisAsyContext);  
		redisAsyncFree(s_redisAsyContext);  
		s_redisAsyContext = NULL;  
	}  

	return true;  
}  

bool subscribe()  
{  
	int ret = redisAsyncCommand(s_redisAsyContext,command_callback, NULL, "PSUBSCRIBE __keyspace@0__:%s", g_htProfile.m_szRedisChannel); 
	if (REDIS_ERR == ret)  
	{  
		LOG__ERROR("Subscribe command failed: %d", ret);

		return false;  
	}  

	LOG__INFO("Subscribe success: __keyspace@0__:%s",g_htProfile.m_szRedisChannel);  
	return true;  
}  

void connect_callback(const redisAsyncContext *s_redisAsyContext,  
		int status)  
{  
	if (status != REDIS_OK)  
	{  
		LOG__ERROR("Error: %s", s_redisAsyContext->errstr);  
	}  
	else  
	{  
		LOG__INFO("Redis Ays connected!");  
	}  
}  




void command_callback(redisAsyncContext *s_redisAsyContext,  
		void *reply, void *privdata) 

{
	char szKey[128] = {0};	
	st_OptionRedis	st_OptRedis = {0};
	if (NULL == reply ) {  
		return ;  
	}  

	redisReply *redis_reply = reinterpret_cast<redisReply *>(reply);  


	if (redis_reply->type == REDIS_REPLY_ARRAY &&  
			redis_reply->elements == 4)  
	{  
		LOG__INFO("redis option:%s--%s",redis_reply->element[3]->str,redis_reply->element[2]->str);
		getParameterValue(redis_reply->element[2]->str, szKey,":");

		strcpy(st_OptRedis.szOption,redis_reply->element[3]->str);
		strcpy(st_OptRedis.szKey,szKey);

		if(!strcmp(st_OptRedis.szOption,"set"))
		{
			redisReply* redisCmdReply = (redisReply*)redisCommand(s_pRedisCon,"get %s",st_OptRedis.szKey);
			string strKey = st_OptRedis.szKey;
			string strValue = redisCmdReply->str;
			pthread_rwlock_wrlock(s_pMiThreadRWLock);
			p_rdisMap->insert(pair<string,string>(strKey,strValue));
			pthread_rwlock_unlock(s_pMiThreadRWLock);            
		}
		else if (!strcmp(st_OptRedis.szOption,"del"))
		{
			string strKey = st_OptRedis.szKey;
			pthread_rwlock_wrlock(s_pMiThreadRWLock);
			p_rdisMap->erase(strKey);
			pthread_rwlock_unlock(s_pMiThreadRWLock);

		}        

	}

	LOG__INFO("map:%d",p_rdisMap->size());
}  


bool eventConnect()  
{  
	// connect redis  
	s_redisAsyContext = redisAsyncConnect(g_htProfile.m_szRedisIp, g_htProfile.m_nPort);  
	if (NULL == s_redisAsyContext)  
	{  
		LOG__ERROR("redisAsy Connect redis failed.");  
		return false;  
	}  

	if (s_redisAsyContext->err)  
	{  
		LOG__ERROR("redisAsy Connect redis error: %d, %s",s_redisAsyContext->err, s_redisAsyContext->errstr);    
		return false;  
	} 

	int ret = redisAsyncCommand(s_redisAsyContext,NULL, NULL,"AUTH %s",g_htProfile.m_szRedisAuth);  
	if (REDIS_ERR == ret)  
	{  
		LOG__ERROR("Ays Auth command failed: %d", ret);

		return false;  
	}  


	// attach the event  
	redisLibeventAttach(s_redisAsyContext, s_eventBase);   

	redisAsyncSetConnectCallback(s_redisAsyContext, &connect_callback);  
	redisAsyncSetDisconnectCallback(s_redisAsyContext, &disconnect_callback);  

	return true;  
} 

void disconnect_callback( const redisAsyncContext *s_redisAsyContext, int status)  
{  
	if (status != REDIS_OK)  
	{   
		LOG__ERROR("Error: %s, again connect", s_redisAsyContext->errstr);
		eventConnect();

	}  
} 

const string LookupMapValue(const char* pKey)
{
	string  strKey(pKey);
	string  strValue = "";

	pthread_rwlock_rdlock(s_pMiThreadRWLock);
	strValue = p_rdisMap->at(strKey);
	pthread_rwlock_unlock(s_pMiThreadRWLock);
	if(strValue == "")
	{
		LOG__WARN("map key:%s value is NULL",pKey);
	}
	return strValue;
} 

